package com.whale.springboot.kafka.java.consumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Properties;

/**
 * 普普通通的消费者
 */
public class CustomConsumer {

    public static void main(String[] args) {
        // 1.创建消费者的配置对象
        Properties properties = new Properties();
        // 2.给消费者配置对象添加参数
        /**
         * 填写kafka服务器，如果是个集群，你这边填一个也行，他内部会自行去找到整个集群所有节点，但是
         * 你如果就写了一个，有个缺点，当你写的这个节点挂了，那真的就找不到大部队了。
         * 故，建议把所有的节点都写进去
         */
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.179.131:9092");
        // 配置序列化 必须
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 配置消费者组（组名任意起名） 必须
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "JKH1");
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true"); //自动提交时异步提交，丢数据&&重复数据
//        properties.setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"15000");// 开启自动提交之后，会在多少秒之后提交offset

        /**
         * 重置消费位置。
         * earliest：从第一个消息开始取
         * latest： 从没有被提交的第一个消息开始取
         *
         * 优先级：如果消费组里面已经有offset的提交记录了，则还是按照offset接着消费，AUTO_OFFSET_RESET_CONFIG会失效，不起作用 （优先级很低）
         * AUTO_OFFSET_RESET_CONFIG 只有在消费组第一次启动的时候才有效感觉，所以才叫他"重置消费位置"
         */
        properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        // 创建消费者对象
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(properties);
        // 注册要消费的主题（可以消费多个主题）
        ArrayList<String> topics = new ArrayList<>();
        topics.add("test-more");

        // 小知识，当添加、删除 消费者的时候，消费组内会触发重平衡。
        kafkaConsumer.subscribe(topics);
//         消费某个主题的某个分区数据
//        ArrayList<TopicPartition> topicPartitions = new ArrayList<>();
//        topicPartitions.add(new TopicPartition("whale", 0));
//        kafkaConsumer.assign(topicPartitions);

        // 拉取数据打印
        while (true) {

            /**
             * poll的含义：
             * 消费者去拉取数据的时候，如果服务端没有数据相应，会保持连接，等待服务器响应
             * poll中传入的参数，就是指  等待的最大时长
             */
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
            System.out.println("当前组的数量：" + consumerRecords.count());
            try {
                // 打印消费到的数据
                for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                    // offset
                    // topic
                    // partition
                    // key
                    // value
                    // timestamp：时间戳
                    // timestampType：时间戳类型，   本条记录的创建时间（生产者）、本条记录的追加时间（broker写入log文件的时间）  2选1

                    // headers：数据头，生产者在写入数据的时候，可以附加加进来的一些额外数据，这是用户自定义的

                    // 假装正常消费
                    System.out.println("offset：" + consumerRecord.offset());
                    System.out.println("value：" + consumerRecord.value());
                    System.out.println("分区：" + consumerRecord.partition());
                    System.out.println("------------------------------------------");
                }
            }
            catch (Exception e) {
                System.out.println("报错了~~~");
            }
        }
    }
}